package com.boot2523.myrabbitmq.q2;

import com.rabbitmq.client.*;

import java.io.IOException;

public class Worker {
    private static final String QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("guest");
        factory.setPassword("guest");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        // 队列声明（与生产者一致）
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        System.out.println(" [*] Waiting for messages...");

        // 设置每个消费者最多同时处理1条消息（公平调度）
//        限制每个消费者在同一时间最多只处理 1 条消息，即“一次只处理一条未确认的消息”。
//        这是 RabbitMQ 中实现负载均衡和防止消费者过载的常用手段。
//        channel.basicQos(int prefetchCount);
//        prefetchCount = 1：表示该通道上的消费者最多同时接收并处理 1 条未确认的消息。
//• RabbitMQ 将不会向这个消费者推送新的消息，直到它确认了当前这条消息。
        channel.basicQos(1);

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            try {
                System.out.println(" [Worker] Received: " + message);
                doWork(message); // 模拟处理任务
                System.out.println(" [Worker] Done: " + message);
            } finally {
                // 手动确认消息
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        };

        // 开启手动确认模式
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
    }

    private static void doWork(String task) {
        // 模拟长时间任务
        for (char ch : task.toCharArray()) {
            if (ch == '.') {
                try {
                    Thread.sleep(1000); // 每个点暂停一秒
                } catch (InterruptedException _ignored) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}
